/*
 * Title:        CloudScheduler Toolkit
 * Description:  a Toolkit for Modeling and Simulation of Job Scheduling and Resource Provisioning in Cloud System
 * Licence:      GPL - http://www.gnu.org/copyleft/gpl.html
 * Author:       Fuhui Wu
 *
 * Copyright (c) 2013-2014, The National University of Defense Technology, China
 */

package org.nudt.jCloud.jobScheduler;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;

import org.nudt.jCloud.infrastructure.Host;
import org.nudt.jCloud.infrastructure.Slot;
import org.nudt.jCloud.simulation.Batch;
import org.nudt.jCloud.simulation.CloudSchedSIM;
import org.nudt.jCloud.simulation.Framework;
import org.nudt.jCloud.simulation.Request;
import org.nudt.jCloud.simulation.RequestData;
import org.nudt.jCloud.simulation.SIMTag;
import org.nudt.jCloud.simulation.ScaleData;
import org.nudt.jCloud.simulation.SchedBatchSet;
import org.nudt.jCloud.simulation.SimEvent;
import org.nudt.jCloud.simulation.User;
import org.nudt.jCloud.simulation.UserBroker;
import org.nudt.jCloud.simulation.VirtSlot;
import org.nudt.jCloud.simulation.VirtSlotAlloc;
import org.nudt.jCloud.util.Log;
import org.nudt.jCloud.workload.Job;
import org.nudt.jCloud.workload.LogicalJob;
import org.nudt.jCloud.workload.Task;

public abstract class JobScheduler {
	private String name;
	private Framework frwk;
	private SortedSet<LogicalJob> submittedLogicalJobs;
	private List<Batch> scheduledBatches;

	private List<VirtSlotAlloc> virtSlotAllocList;

	public JobScheduler(String name) {
		setName(name);
		submittedLogicalJobs = new TreeSet<LogicalJob>();
		scheduledBatches = new ArrayList<Batch>();
		virtSlotAllocList = new ArrayList<VirtSlotAlloc>();
	}

	public void reset() {
		setFrwk(null);
		submittedLogicalJobs.clear();
		scheduledBatches.removeAll(scheduledBatches);
	}

	public void nextEvent() {
		Collections.sort(scheduledBatches, new SortBatchByFinishTime());
		double updateTime = Double.MAX_VALUE;
		if (scheduledBatches.size() > 0) {
			updateTime = scheduledBatches.get(0).getFinishTime();
		}
		double submitTime = earliestJob();
		double nextTime = Math.min(updateTime, submitTime);
		if (getFrwk().getUserBroker().hasUnSubmittedXJob()
				|| scheduledBatches.size() > 0) {
			getFrwk().getUserBroker().send(getFrwk().getUserBroker().getId(),
					nextTime - CloudSchedSIM.getClock(),
					SIMTag.UPDATE_OR_SUBMIT_JOB, nextTime);
		}
	}

	public double earliestJob() {
		double earliestJob = Double.MAX_VALUE;
		for (int i = 0; i < getFrwk().getUserBroker().getUserList().size(); ++i) {
			User user = getFrwk().getUserBroker().getUserList().get(i);
			if (user.getSubmittedXJobCount() < user.getWl().size()) {
				double arrival = user.getWl().getLogicalJobList()
						.get(user.getSubmittedXJobCount()).getArrivalTime();
				if (arrival < earliestJob) {
					earliestJob = arrival;
				}
			}
		}
		return earliestJob;
	}

	/**
	 * param "nextTime" is received from the nextEvent() function. At the time
	 * of "nextTime", there may batches finish, new job be submitted or the
	 * both. If any situation happens, job scheduling is triggered.
	 * 
	 * @param nextTime
	 */
	public void processNextEvent(double nextTime) {
		updateBatch(nextTime);
		submitXJob(nextTime);
		RequestData requestData = new RequestData();
		requestData.init(request(), getFrwk().getUserBroker().getRM_Mode());
		if (requestData.getRequestList().size() > 0) {
			switch (getFrwk().getUserBroker().getRM_Mode()) {
			case SIMTag.RM_MODE_R:
				ScaleData scaleData = new ScaleData(getFrwk().getUserBroker().getId(), requestData);
				getFrwk().getResourceScaler().scaleResource(requestData, scaleData);
				getFrwk().getUserBroker().sendNow(	getFrwk().getResManager().getId(),
						SIMTag.SCALE_RESOURCE, scaleData);
				break;
			case SIMTag.RM_MODE_S:
				scheduleJob(requestData);
				break;
			default:
				throw new RuntimeException("Unsupported resource utilize mode.");
			}
		} else {
			nextEvent();
		}
	}

	public void processResCommitACK(VirtSlotAlloc virtSlotAlloc) {
		Log.frwkLog(CloudSchedSIM.getClock(), getFrwk());
		Map<Host, List<VirtSlot>> committedVirtSlotSet = virtSlotAlloc.getVirtSlotSet();
		SchedBatchSet schedBatchSet = new SchedBatchSet();
		int concurrentFailure = 0;
		for(Host host : committedVirtSlotSet.keySet()){
			List<VirtSlot> virtSlotList = committedVirtSlotSet.get(host);
			for(int i=0; i<virtSlotList.size(); ++i){
				VirtSlot virtSlot = virtSlotList.get(i);
				if(virtSlot.getSlot() != null){
					getFrwk().leaseSlot(virtSlot.getSlot());
					getFrwk().busySlot(virtSlot.getSlot());
					virtSlot.getTask().setSlot(virtSlot.getSlot());
					schedBatchSet.add(virtSlot.getTask());
				}else{
					concurrentFailure = concurrentFailure+1;
					virtSlot.getTask().cancelAssignToSlot();
				}
			}
		}
		if(concurrentFailure>0){
			Log.concurrentFailureLog(getFrwk().getUserBroker().getName(), CloudSchedSIM.getClock(), concurrentFailure);
		}
		virtSlotAllocList.remove(virtSlotAlloc);
		scheduleBatch(schedBatchSet);
		UserBroker userBroker = getFrwk().getUserBroker();
		List<SimEvent> events = CloudSchedSIM.getFuture().get(userBroker.getId(), SIMTag.UPDATE_OR_SUBMIT_JOB);
		for(int j=0; j<events.size(); ++j){
			CloudSchedSIM.getFuture().remove(events.get(j));
		}
		nextEvent();
	}

	public void scheduleJob(RequestData requestData) {
		switch (getFrwk().getUserBroker().getRM_Mode()) {
		case SIMTag.RM_MODE_R:
			for(Framework frwk : getFrwk().getResManager().getFrwkList()){
				Log.frwkLog(CloudSchedSIM.getClock(), frwk);
			}
			SchedBatchSet schedBatchSet = new SchedBatchSet();
			assignTaskR(requestData, schedBatchSet);
			scheduleBatch(schedBatchSet);
			nextEvent();
			break;
		case SIMTag.RM_MODE_S:
			VirtSlotAlloc virtSlotAlloc = new VirtSlotAlloc(getFrwk());
			assignTaskS(requestData, virtSlotAlloc);
			if (virtSlotAlloc.isEmpty() == false) {
				virtSlotAllocList.add(virtSlotAlloc);
				double delay = computeDelay(requestData);
				Log.jobSchedulerDelayLog(getFrwk().getUserBroker().getName(),
						CloudSchedSIM.getClock(), delay);
				getFrwk().getUserBroker().send(
						getFrwk().getResManager().getId(), delay,
						SIMTag.RESOURCE_COMMIT, virtSlotAlloc);
			}
			nextEvent();
			break;
		default:
			throw new RuntimeException("Unsupported resource utilize mode.");
		}
	}
	
	public double computeDelay(RequestData requestData) {
		double delay = 0;
		for (int i = 0; i < requestData.getRequestList().size(); ++i) {
			Request request = requestData.getRequestList().get(i);
			delay += request.getJob().getXJob().getWl().getThinkTimeOfJob()+
					request.getJob().getXJob().getWl().getThinkTimeOfTask()*(request.getAssignedTasks().size());
		}
		return delay;
	}

	/*
	 * Submit next job at nextTime.
	 */
	public void submitXJob(double nextTime) {
		for (int i = 0; i < getFrwk().getUserBroker().getUserList().size(); ++i) {
			User user = getFrwk().getUserBroker().getUserList().get(i);
			if (user.getSubmittedXJobCount() < user.getWl().size()) {
				while (nextTime == user.getWl().getLogicalJobList()
						.get(user.getSubmittedXJobCount()).getArrivalTime()) {
					submittedLogicalJobs.add(user.getWl().getLogicalJobList()
							.get(user.getSubmittedXJobCount()));
					user.getWl().getLogicalJobList()
							.get(user.getSubmittedXJobCount()).submit();
					user.setSubmittedXJobCount(user.getSubmittedXJobCount() + 1);
					if (user.getSubmittedXJobCount() >= getFrwk()
							.getUserBroker().getUserList().get(0).getWl()
							.size()) {
						break;
					}
				}
			}
		}
	}

	/**
	 * After the job scheduling, some tasks from the request are selected and
	 * assigned to slots. These tasks are organized into batches according to
	 * job scheduler strategy. For each batch, the tasks are selected from the
	 * same job. But all tasks of the same job may be scheduled into multiple
	 * batches, because of computing capability or data locality factors. In the
	 * process, the job type is checked if a batch comes from a job of
	 * "all-or-nothing" job transaction mode. In customized job scheduler,
	 * however, it is the responsibility of "assignTask()" to address this
	 * problem.
	 * 
	 * @param schedBatchSet
	 */
	public void scheduleBatch(SchedBatchSet schedBatchSet) {
		List<User> userList = getFrwk().getUserBroker().getUserList();
		for (User user : userList) {
			Map<Job, Map<Double, Batch>> jobBatchList = schedBatchSet
					.getBatchSet().get(user);
			if (jobBatchList != null) {
				for (Job job : jobBatchList.keySet()) {
					Map<Double, Batch> durationBatches = jobBatchList.get(job);
					for (Double duration : durationBatches.keySet()) {
						Batch batch = durationBatches.get(duration);
						if (batch.getJob().getType() == "GANG"
								&& batch.size() < batch.getJob().size()) {
							cancelSchedBatch(batch);
						} else {
							batch.setScheduleTime(CloudSchedSIM.getClock());
							batch.setFinishTime(CloudSchedSIM.getClock()
									+ durationBatches.get(duration)
											.getDuration());
							job.scheduleBatch(batch);
							user.getUserBroker().getFrwk().getJobScheduler()
									.getScheduledBatches().add(batch);
						}
					}
				}
			}
		}
	}

	/*
	 * Update job state at nextTime.
	 */
	public void updateBatch(double nextTime) {
		Collections.sort(scheduledBatches, new SortBatchByFinishTime());
		int count = scheduledBatches.size();
		for (int i = 0; i < count; ++i) {
			Batch batch = scheduledBatches.get(0);
			if (batch.getFinishTime() > nextTime) {// no batch to finish.
				break;
			} else {
				// free the slots occupied by current batch firstly.
				for (int j = 0; j < batch.size(); ++j) {
					getFrwk().freeSlot(batch.getTaskList().get(j).getSlot());
				}
				switch (getFrwk().getUserBroker().getRM_Mode()) {
				case SIMTag.RM_MODE_R:
					break;
				case SIMTag.RM_MODE_S:
					for (int j = 0; j < batch.size(); ++j) {
						Slot slot = batch.getTaskList().get(j).getSlot();
						getFrwk().releaseFreeSlot(slot);
					}
					break;
				default:
					throw new RuntimeException("Unsupported resource manager.");
				}
				scheduledBatches.get(0).finish();
				if (scheduledBatches.get(0).getJob().getXJob().isFinished()) {
					submittedLogicalJobs.remove(scheduledBatches.get(0).getJob()
							.getXJob());
				}
				scheduledBatches.remove(0);
			}
		}
	}

	public List<Request> request() {
		List<Request> requests = new ArrayList<Request>();
		for(LogicalJob logicalJob : getSubmittedLogicalJobs()){
			requests.addAll(logicalJob.request());
		}
		return requests;
	}

	public abstract void assignTaskR(RequestData requestData,
			SchedBatchSet schedBatchSet);

	/**
	 * There may be more than one VirtSlotAlloc in the same job scheduler, so it
	 * is necessary to consider the allocated VirtSlot in the virtSlotAllocList.
	 * 
	 * @param requestData
	 * @param virtSlotAlloc
	 */
	public abstract void assignTaskS(RequestData requestData,
			VirtSlotAlloc virtSlotAlloc);

	public void assignTasktoSlot(Request request, Task task, Slot slot,
			SchedBatchSet schedBatchSet) {
		task.assignToSlot(slot);
		request.assignTask(task);
		getFrwk().busySlot(slot);
		schedBatchSet.add(task);
	}

	public void assignTaskToVirtSlot(Request request, Task task,
			VirtSlot virtSlot) {
		virtSlot.setTask(task);
		task.assignToVirtSlot(virtSlot);
		request.assignTask(task);
	}

	public void cancelRequestR(Request request, SchedBatchSet schedBatchSet) {
		int assignedTaskNum = request.getAssignedTasks().size();
		for (int i = 0; i < assignedTaskNum; ++i) {
			Task task = request.getAssignedTasks().get(0);
			task.cancelAssignToSlot();
			schedBatchSet.removeTask(task);
			Slot slot = task.getSlot();
			getFrwk().freeSlot(slot);
			request.cancelTaskAssign(task);
		}
	}
	
	public void cancelRequestS(Request request, VirtSlotAlloc virtSlotAlloc) {
		int assignedTaskNum = request.getAssignedTasks().size();
		for(int i=0; i<assignedTaskNum; ++i){
			Task task = request.getAssignedTasks().get(0);
			task.cancelAssignToVirtSlot();
			request.cancelTaskAssign(task);
			virtSlotAlloc.removeVirtSlot(task);
		}		
	}

	public void cancelSchedBatch(Batch batch) {
		for (Task task : batch.getTaskList()) {
			task.cancelAssignToSlot();
			Slot slot = task.getSlot();
			switch (getFrwk().getUserBroker().getRM_Mode()) {
			case SIMTag.RM_MODE_R:
				getFrwk().freeSlot(slot);
				break;
			case SIMTag.RM_MODE_S:
				getFrwk().releaseBusySlot(slot);
				break;
			default:
				throw new RuntimeException("Unsupported resource utilize mode.");
			}
		}
	}	
	
	class SortBatchByFinishTime implements Comparator<Batch> {
		public int compare(Batch batch1, Batch batch2) {
			if (batch1.getFinishTime() > batch2.getFinishTime()) {
				return 1;
			} else if (batch1.getFinishTime() < batch2.getFinishTime()) {
				return -1;
			} else {
				return 0;
			}
		}
	}
	
	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public Framework getFrwk() {
		return frwk;
	}

	public void setFrwk(Framework frwk) {
		this.frwk = frwk;
	}

	public List<Batch> getScheduledBatches() {
		return scheduledBatches;
	}

	public void setScheduledBatches(List<Batch> scheduledBatches) {
		this.scheduledBatches = scheduledBatches;
	}

	public List<VirtSlotAlloc> getVirtSlotAllocList() {
		return virtSlotAllocList;
	}

	public void setVirtSlotAllocList(List<VirtSlotAlloc> virtSlotAllocList) {
		this.virtSlotAllocList = virtSlotAllocList;
	}

	public void setSubmittedLogicalJobs(SortedSet<LogicalJob> submittedLogicalJobs) {
		this.submittedLogicalJobs = submittedLogicalJobs;
	}
	
	public SortedSet<LogicalJob> getSubmittedLogicalJobs(){
		return submittedLogicalJobs;
	}
	
}
